package cn.restcloud.example.controller;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;

import cn.restcloud.mqbus.util.kafka.KafkaConsumerUtils;
import cn.restcloud.mqbus.util.kafka.KafkaProducerUtils;
import org.apache.kafka.clients.producer.RecordMetadata;
import cn.restcloud.framework.core.annotation.bean.BeanConfig;
import cn.restcloud.framework.core.annotation.bean.BeanType;
import cn.restcloud.framework.core.annotation.rest.MethodType;
import cn.restcloud.framework.core.annotation.rest.RestConfig;
import cn.restcloud.framework.core.util.DateTimeUtil;
import cn.restcloud.framework.view.base.DocAndView;
import cn.restcloud.framework.view.util.ViewUtil;

@BeanConfig(appId = "example",beanId = "KafkaTestAPI",beanName = "kafka消息示例",beanType = BeanType.CONTROLLER)
@RestConfig(url = "/rest/example/kafka")
public class KafkaTestAPI {	
	
	@RestConfig(configName = "发送kafka消息",method = MethodType.GET,url = "/send")
    public DocAndView sendMessage() throws Exception{
    	String msg="来自kafka TestMessage="+DateTimeUtil.getNow();
    	Future<RecordMetadata> metadataFuture = KafkaProducerUtils.getInstance("ecskafka").sendOneMessage("RestCloud", msg);
    	//同步获得 Future 对象的结果
        RecordMetadata recordMetadata = metadataFuture.get();
        String sendMsg=String.format("Produce ok: topic:%s partition:%d offset:%d value:%s",recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset(), msg);
    	return ViewUtil.getSuccessView(sendMsg);
    	
    }
    
	@RestConfig(configName = "接收kafka消息",method = MethodType.GET,url = "/receive")
    public DocAndView reciveMessage() throws Exception{
		List<String> topics=new ArrayList<String>();
		topics.add("RestCloud");
		KafkaConsumerUtils kafkaConsumerUtils= KafkaConsumerUtils.getInstance("ecskafkaclient",topics);
		String msg=kafkaConsumerUtils.getOneKafkaData();
    	return ViewUtil.getJsonView(msg);
    }
	
}
